#include "zhelpers.h"
#include "proto.msg.pb.h"
#include "lf_thread.h"
#include "lf_server.h"
#include "Log.h"
#include "vs_opencv.h"
#include "LocalFeaturesSearch.h"

int lf_match_thread::thread_func(){
	if(get_thread_pool() == NULL){
		LOG_ERROR_VA("[%s] pool_work_thread don't belong to a thread_pool", get_name());
		return -1;
	}
	int ret;
	void *receiver = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REP);
	if(receiver == NULL){
		LOG_ERROR_VA("[%s] zmq_socket failed", get_name());
		return -1;
	}
	//ret = zmq_connect(receiver, "inproc://workers");
	ret = zmq_connect(receiver, "tcp://199.155.122.11:45516"); //lf_match back_address

	LOG_DEBUG_VA("[%s] thread start...", get_name());
	while(!(get_thread_pool()->has_stoped())){
		int size;
		zmq_msg_t recv_msg;
		zmq_msg_init(&recv_msg);

		LOG_DEBUG_VA("[%s] thread get task...", get_name());
		size = zmq_recvmsg(receiver, &recv_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		string proto_buf;
		proto_buf.assign((char *)zmq_msg_data(&recv_msg), size);
		proto::lf_match_req lf_match_req;
		if(!lf_match_req.ParseFromString(proto_buf)){
			LOG_ERROR_VA("[%s] lf_match_req ParseFromString fail", get_name());
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("[%s] recv length: %d", get_name(), size);
		//start Match
		database_bdb *db = get_database();

		int vid_num = lf_match_req.vid_num();
		int *vids = new int[vid_num];
		LocalFeatures::str_Feature *feature_fv = (LocalFeatures::str_Feature *)lf_match_req.mutable_feature_data()->c_str();
		for(int i = 0; i < vid_num; i++){
			vids[i] = feature_fv[i].word;
		}
		vid_data *vid_datas = new vid_data[vid_num];
		if (db->get_data(lf_match_req.class_id(), lf_match_req.b2c_flag(), database::lf, vid_num, vids, vid_datas)  <= 0) {
			LOG_ERROR_VA("[lf_match] db get data failed, check data_root_path config");
			send_error_reply(receiver);
			delete [] vid_datas;
			delete [] vids;
			continue;
		}
		int total_word_count = db->get_total_word_count();

		LocalFeatures::strDataStruct *datas = new LocalFeatures::strDataStruct[vid_num];
		for (int i = 0; i != vid_num; i++) {
			datas[i].dataLen = vid_datas[i].length;
			datas[i].data = vid_datas[i].data;
		}
		delete [] vid_datas;
		int res_num = lf_match_req.res_num();
		LocalFeatures::strImageRes *res = new LocalFeatures::strImageRes[res_num];

		ret = LocalFeatures::Match(total_word_count, lf_match_req.class_id(), 0, vid_num, datas, feature_fv, res_num, res, lf_match_req.merge_same()); 
		if(ret != 0){
			LOG_ERROR_VA("LocalFeatures::Match failed: %d", ret);
			send_error_reply(receiver);
			for (int i = 0; i != vid_num; i++)
				if (datas[i].dataLen)
					delete [] datas[i].data;
			delete [] datas;
			delete [] res;
			continue;
		}
		for (int i = 0; i != vid_num; i++)
			if (datas[i].dataLen)
				delete [] datas[i].data;
		delete [] datas;
		//end of match
		//result in GLOBALFEATURE_3_2_0::TDistributedResult res;
		proto::lf_match_resp lf_match_resp;
		lf_match_resp.mutable_head()->set_version(420);
		lf_match_resp.mutable_head()->set_command(420);
		lf_match_resp.mutable_head()->set_password("xxxxxxxx", 8);
		lf_match_resp.mutable_head()->set_type(1);
		lf_match_resp.mutable_head()->set_status(0);
		lf_match_resp.mutable_head()->set_src(1);
		lf_match_resp.mutable_head()->set_dest(2);
		lf_match_resp.mutable_head()->set_sn_high(123456789);
		lf_match_resp.mutable_head()->set_sn_low(123456789);
		lf_match_resp.mutable_head()->set_length(0);

		lf_match_resp.set_str_num(res_num);
		lf_match_resp.set_str(res, sizeof(LocalFeatures::strImageRes)*res_num);

		LOG_DEBUG_VA("[%s] lf_match_resp str_num: %d", get_name(),lf_match_resp.str_num());
		delete [] res;

		proto_buf.clear();
		if(!lf_match_resp.SerializePartialToString(&proto_buf)){
			LOG_ERROR_VA("SerializePartialToString fail");
			send_error_reply(receiver);
			continue;
		}

		zmq_msg_t lf_match_resp_msg;
		zmq_msg_init_size(&lf_match_resp_msg, proto_buf.size());
		memcpy(zmq_msg_data(&lf_match_resp_msg), proto_buf.c_str(), proto_buf.size());
		size = zmq_sendmsg(receiver, &lf_match_resp_msg, 0);
		if(size == -1){
			send_error_reply(receiver);
			LOG_ERROR_VA("zmq_sendmsg error");
			continue;
		}
	}
	zmq_close(receiver);
}

int lf_merge_thread::thread_func(){
	if(get_thread_pool() == NULL){
		LOG_ERROR_VA("[%s] pool_work_thread don't belong to a thread_pool", get_name());
		return -1;
	}
	int ret;
	void *receiver = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REP);
	if(receiver == NULL){
		LOG_ERROR_VA("[%s] zmq_socket failed", get_name());
		return -1;
	}
	ret = zmq_connect(receiver, "tcp://199.155.122.11:45515"); //lf_merge back_address

	LOG_DEBUG_VA("[%s] thread start...", get_name());
	while(!(get_thread_pool()->has_stoped())){
		int size;
		zmq_msg_t recv_msg;
		zmq_msg_init(&recv_msg);

		LOG_DEBUG_VA("[%s] thread get task...", get_name());
		size = zmq_recvmsg(receiver, &recv_msg, 0);
		if(size == -1){
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		string proto_buf;
		proto_buf.assign((char *)zmq_msg_data(&recv_msg), size);
		proto::lf_match_req lf_match_req;
		if(!lf_match_req.ParseFromString(proto_buf)){
			LOG_ERROR_VA("[%s] lf_match_req ParseFromString fail", get_name());
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("[%s] recv length: %d", get_name(), size);
		//send to all match
		int send_match_req_num = 0;
		string lf_match_tcp_addr;
		vector<string> alive_lf_match_addr_vec;
		vector<void *> alive_lf_match_vec;
		for(vector<address>::iterator it = get_lf_match().begin(); it != get_lf_match().end(); it++){
			//connect every time
			//not a good idea
			lf_match_tcp_addr.clear();
			lf_match_tcp_addr.append("tcp://");
			lf_match_tcp_addr.append(it->get_ip());
			lf_match_tcp_addr.append(":");
			lf_match_tcp_addr.append(it->get_port());
			void *zmq_lf_match = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REQ);
			if(zmq_lf_match == NULL){
				LOG_ERROR_VA("[%s] match_zmq zmq_socket failed", get_name());
				return -1;
			}
			ret = zmq_connect(zmq_lf_match, lf_match_tcp_addr.c_str()); //lf_match front_address
			if(ret == -1){
				LOG_ERROR_VA("zmq_connect error");
				continue;
			}
			zmq_msg_t match_req_msg;
			zmq_msg_init_size(&match_req_msg, size);
			memcpy(zmq_msg_data(&match_req_msg), zmq_msg_data(&recv_msg), size);
			//zmq_msg_copy(&match_req_msg, &recv_msg);
			int size = zmq_sendmsg(zmq_lf_match, &match_req_msg, 0);
			if(size == -1){
				LOG_ERROR_VA("zmq_sendmsg error");
				continue;
			}
			alive_lf_match_vec.push_back(zmq_lf_match);
			send_match_req_num++;
		}
		//send over, start to receive for all matchs
		if(send_match_req_num == 0){
			LOG_ERROR_VA("no match send success");
			send_error_reply(receiver);
			continue;
		}
		//zmq_poll recv
		int recv_match_req_num = 0;
		vector<zmq_msg_t> lf_match_resp_msg_vec;
		vector<proto::lf_match_resp> lf_match_resp_vec;
		while(recv_match_req_num < send_match_req_num){
			//the timeout!!!
			LOG_DEBUG_VA("[%s] prepare to recv from %d match", get_name(), send_match_req_num);
			//zmq_pollitem_t *poll_items = new zmq_pollitem_t[alive_lf_match_vec.size()];
			zmq_pollitem_t poll_items[alive_lf_match_vec.size()];
			for(int i = 0; i < alive_lf_match_vec.size(); i++){
				poll_items[i].socket = alive_lf_match_vec[i];
				poll_items[i].fd = 0;
				poll_items[i].events = ZMQ_POLLIN;
				poll_items[i].revents = 0;
			}
			int zmq_poll_ret = zmq_poll(poll_items, alive_lf_match_vec.size(), -1);
			if(zmq_poll_ret < 0)
				LOG_ERROR_VA("[%s] zmq_poll error");
			int index = 0;
			int alive_num = alive_lf_match_vec.size();
			for(int i = 0; i < alive_num; i++){
				if(poll_items[i].revents & ZMQ_POLLIN){
					LOG_DEBUG_VA("[%s] get a ZMQ_POLLIN", get_name());
					zmq_msg_t lf_match_resp_msg;
					zmq_msg_init(&lf_match_resp_msg);
					zmq_msg_recv(&lf_match_resp_msg, poll_items[i].socket, 0);
					//if recv sucess, erase if from alive_lf_match_vec!!!
					proto_buf.clear();
					proto_buf.assign((char *)zmq_msg_data(&lf_match_resp_msg),zmq_msg_size(&lf_match_resp_msg));
					proto::lf_match_resp lf_match_resp;
					if(!lf_match_resp.ParseFromString(proto_buf))
						LOG_ERROR_VA("[%s] lf_match_resp ParseFromString fail", get_name());
					else
						lf_match_resp_vec.push_back(lf_match_resp);
					alive_lf_match_vec.erase(alive_lf_match_vec.begin() + index);
					index--;
					recv_match_req_num++;
					zmq_msg_close(&lf_match_resp_msg);
				}
				index++;
			}
		}
		//recv all match result sucess
		//result in lf_match_resp_msg_vec
		//start Merge
		
		int resp_num = lf_match_resp_vec.size();
		LOG_DEBUG_VA("[%s] recv %d lf_match_resp", get_name(), resp_num);
		const LocalFeatures::strImageRes **in_res = new const LocalFeatures::strImageRes* [resp_num];
		int *in_res_num = new int[resp_num];

		for(int i = 0; i < resp_num; i++){
			//the input of merge
			in_res_num[i] = lf_match_resp_vec[i].str_num();
			in_res[i] = (LocalFeatures::strImageRes *)(lf_match_resp_vec[i].mutable_str()->c_str());
		}//end of lf_match_resp_vec

		//output of merge
		int res_num = lf_match_req.res_num();
		LocalFeatures::strImageRes *res = new LocalFeatures::strImageRes[res_num];

		ret = LocalFeatures::Merge(resp_num, in_res_num, in_res, res_num, res, lf_match_req.merge_same());
		if(ret != 0){
			LOG_ERROR_VA("[%s] Merge failed: %d", get_name(), ret);
			send_error_reply(receiver);
			delete [] in_res_num;
			delete [] in_res;
			delete [] res;
			continue;
		}
		delete [] in_res_num;
		delete [] in_res;
		//end of merge
		//start send resp to front
		LOG_DEBUG_VA("[%s] prepare to send lf_merge_resp to front", get_name());
		proto::lf_merge_resp lf_merge_resp;
		lf_merge_resp.mutable_head()->set_version(420);
		lf_merge_resp.mutable_head()->set_command(420);
		lf_merge_resp.mutable_head()->set_password("xxxxxxxx", 8);
		lf_merge_resp.mutable_head()->set_type(1);
		lf_merge_resp.mutable_head()->set_status(0);
		lf_merge_resp.mutable_head()->set_src(1);
		lf_merge_resp.mutable_head()->set_dest(2);
		lf_merge_resp.mutable_head()->set_sn_high(123456789);
		lf_merge_resp.mutable_head()->set_sn_low(123456789);
		lf_merge_resp.mutable_head()->set_length(0);

		lf_merge_resp.set_res_num(res_num);
		for(int i = 0; i < res_num; i++){
			lf_merge_resp.add_iid(res[i].imageID);
			lf_merge_resp.add_gid(res[i].groupID);
			lf_merge_resp.add_simi(res[i].similarity);
			lf_merge_resp.add_type(res[i].imageType);
			lf_merge_resp.add_source(res[i].source);
		}
		
		delete [] res;
		LOG_DEBUG_VA("[%s] lf_merge_resp res_num: %d", get_name(),lf_merge_resp.res_num());
		for(int i = 0; i < 5; i++){
			LOG_DEBUG_VA("[%s] merge result iid: %llu, gid: %llu, simi: %d, source: %d", get_name(),
				res[i].imageID, res[i].groupID, res[i].similarity, res[i].source);
		}

		//merge over
		proto_buf.clear();
		if(!lf_merge_resp.SerializePartialToString(&proto_buf)){
			LOG_ERROR_VA("SerializePartialToString fail");
			send_error_reply(receiver);
			continue;
		}
		zmq_msg_t lf_merge_resp_msg;
		zmq_msg_init_size(&lf_merge_resp_msg, proto_buf.size());
		memcpy(zmq_msg_data(&lf_merge_resp_msg), proto_buf.c_str(), proto_buf.size());
		LOG_DEBUG_VA("[%s] send lf_merge_resp to front", get_name());
		size = zmq_sendmsg(receiver, &lf_merge_resp_msg, 0);
		if(size == -1){
			send_error_reply(receiver);
			LOG_ERROR_VA("zmq_sendmsg error");
			continue;
		}
	}
	zmq_close(receiver);
}


int lf_extract_thread::thread_func()
{
	if(get_thread_pool() == NULL){
		LOG_ERROR_VA("[%s] pool_work_thread don't belong to a thread_pool", get_name());
		return -1;
	}
	int ret;
	void *receiver = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REP);
	if(receiver == NULL){
		LOG_ERROR_VA("[%s] zmq_socket failed", get_name());
		return -1;
	}
	//ret = zmq_connect(receiver, "inproc://workers");
	ret = zmq_connect(receiver, "tcp://199.155.122.11:45514"); //gf_extract back_address

	LOG_DEBUG_VA("[%s] thread start", get_name());
	while(!(get_thread_pool()->has_stoped())){
		int size;
		zmq_msg_t recv_msg;
		zmq_msg_init(&recv_msg);

		LOG_DEBUG_VA("[%s] thread get task...", get_name());
		size = zmq_recvmsg(receiver, &recv_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		string proto_buf;
		proto_buf.assign((char *)zmq_msg_data(&recv_msg), size);
		proto::extract_req extract_req;
		if(!extract_req.ParseFromString(proto_buf)){
			LOG_ERROR_VA("[%s] extract_req ParseFromString fail", get_name());
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("[%s] thread recv length: %d", get_name(), size);
		//the work of engine
		//LOG_DEBUG_VA("image width: %d, height: %d, depth: %d, channel: %d, size: %d", extract_req.image_width(), extract_req.image_height(), extract_req.image_depth(), extract_req.image_channel(), extract_req.image_data().size());
		char *image_data = new char[extract_req.image_data().size()];
		memcpy(image_data, extract_req.image_data().c_str(), extract_req.image_data().size());
		IplImage *ipl = cv_create_image(extract_req.image_width(), extract_req.image_height(), extract_req.image_depth(), extract_req.image_channel(), image_data, extract_req.image_data().size());
		delete [] image_data;
		if(ipl == NULL){
			LOG_ERROR_VA("[%d] open_cv load error", get_name());
			send_error_reply(receiver);
			continue;
		}
		
		int vid_num = 500;
		LocalFeatures::str_Feature feature_buf[vid_num];

		ret = LocalFeatures::Extraction(ipl, NULL, 0, NULL, extract_req.class_id(), &vid_num, feature_buf);
		cv_release_image(ipl);
		if(ret != 0){
			LOG_ERROR_VA("LocalFeatures::Extraction failed: %d", ret);
			send_error_reply(receiver);
			continue;
		}

		//send resp to front
		proto::lf_extract_resp lf_extract_resp;
		lf_extract_resp.mutable_head()->set_version(420);
		lf_extract_resp.mutable_head()->set_command(420);
		lf_extract_resp.mutable_head()->set_password("xxxxxxxx", 8);
		lf_extract_resp.mutable_head()->set_type(1);
		lf_extract_resp.mutable_head()->set_status(0);
		lf_extract_resp.mutable_head()->set_src(1);
		lf_extract_resp.mutable_head()->set_dest(2);
		lf_extract_resp.mutable_head()->set_sn_high(123456789);
		lf_extract_resp.mutable_head()->set_sn_low(123456789);
		lf_extract_resp.mutable_head()->set_length(0);

		lf_extract_resp.set_vid_num(vid_num);
		lf_extract_resp.set_feature_data(feature_buf, sizeof(LocalFeatures::str_Feature) * vid_num);
		
		proto_buf.clear();
		if(!lf_extract_resp.SerializePartialToString(&proto_buf)){
			LOG_ERROR_VA("SerializePartialToString fail");
			send_error_reply(receiver);
			continue;
		}
		zmq_msg_t lf_extract_resp_msg;
		zmq_msg_init_size(&lf_extract_resp_msg, proto_buf.size());
		memcpy(zmq_msg_data(&lf_extract_resp_msg), proto_buf.c_str(), proto_buf.size());
		size = zmq_sendmsg(receiver, &lf_extract_resp_msg, 0);
		if(size == -1){
			send_error_reply(receiver);
			LOG_ERROR_VA("zmq_sendmsg error");
			continue;
		}
	}
	zmq_close(receiver);
}

database_bdb *lf_match_thread::get_database(){
	match_server *server =dynamic_cast<match_server *>(get_thread_pool()->get_server());
	return server->get_database();
}

vector<address> &lf_merge_thread::get_lf_match(){
	lf_merge_server *server =dynamic_cast<lf_merge_server *>(get_thread_pool()->get_server());
	return server->get_lf_match();
}

